<?php
/**
 * Created by PhpStorm.
 * User: mac
 * Date: 2019-03-03
 * Time: 00:37
 */

/**
 * Class queue
 * 在子进程里使用队列，代码逻辑上比使用管道要复杂一些
 */
class queue
{
    public $work_num = 0;
    public $work_array = array();
    public $task_data = array();
    public $index = 0;

    public function __construct(array $task_data)
    {
        try {
            $this->work_num = count($task_data);
            $this->task_data = $task_data;
            $this->createProcess();
            $this->pushQueMsg();
            $this->waitProcess();
        } catch (Exception $exception) {
            echo $exception->getMessage();
        }
    }

    private function createProcess()
    {
        for ($i = 0; $i < $this->work_num; $i++) {
            if ($i > 0)$this->index = $i;
            //这里第二个参数设置成了false，表示子进程输出信息不写入管道
            $process = new swoole_process([$this, 'callbackFunc'], false, false);
            //启用消息队列作为进程间通信。这一步会优先于上边的callbackFunc执行。
            $process->useQueue();
            $pid = $process->start();
            $this->work_array[$pid] = $process;
        }
    }

    /**
     * @param swoole_process $work
     * 这个回调函数不会立即执行，而是先保存这样一种状态，等走完下边的程序在执行回调
     */
    public function callbackFunc(swoole_process $work)
    {
        //$process对象并未执行start，也可以执行push/pop向队列推送/提取数据
        $recv = $work->pop();
        echo 'receive message ' . $recv . PHP_EOL;
        $num = $this->getFunc();
        echo $num . ' finish task ' . $this->task_data[$this->index] . PHP_EOL;
        $work->exit(0);
    }

    private function getFunc()
    {
        sleep(3);
        return mt_rand(100, 999);
    }

    public function pushQueMsg()
    {
        $i = 0;
        foreach ($this->work_array as $process) {
            $process->push($this->task_data[$i]);
            $i++;
        }
    }

    private function waitProcess()
    {
        for ($i = 0; $i < $this->work_num; $i++) {
            $ret = swoole_process::wait();
            echo $ret['pid'] . 'exit' . PHP_EOL;
        }
    }
}
$task_data = array('zhangsan', 'lisi', 'wangwu', 'zhaoliu', 'houqi');
$queue = new queue($task_data);
$queue->pushQueMsg();